package cn.com.ytst.data.integration.core.container;

import java.io.File;
import java.util.concurrent.atomic.AtomicBoolean;
import com.alibaba.datax.common.util.Configuration;
import cn.com.ytst.data.integration.core.channel.EventPipleChannel;
import cn.com.ytst.data.integration.core.channel.PipleChannel;
import cn.com.ytst.data.integration.core.plugin.Reader;
import cn.com.ytst.data.integration.core.plugin.Writer;
import cn.com.ytst.data.integration.util.ClassUtil;
import cn.com.ytst.data.integration.util.Constants;

/**
 * @ClassName JobContainer
 * @Deacription
 * @Author Administrator
 * @Date 2020/1/4 0004 15:03
 * @Version 1.0
 **/
public class JobContainer extends AbstractContainer {

    private File f;

    private Reader readerJob;

    private Writer writerJob;


    Configuration configuration;
    PipleChannel pipleChannel;

    private AtomicBoolean writeflage = new AtomicBoolean(false);

    public JobContainer() {
        pipleChannel = new EventPipleChannel();
    }

    @Override
    public void setFile(File file) {
        this.f = file;
        configuration = Configuration.from(f);
    }

    @Override
    public void start() {
        String readClassname = configuration.getString(Constants.READER_PLUGIN_CLASS_TAG);
        //根据这个插件名字找到对应的插件对象
        String writeClassname = configuration.getString(Constants.WRITE_PLUGIN_CLASS_TAG);
        readerinit(readClassname);
        writeinit(writeClassname);
        builderReaderJob();
        builderWriterJob();
    }

    private void readerinit(String readClassname){
        readerJob = ClassUtil.instantiate(readClassname, Reader.class, new Object[]{});
        readerJob.setPipleChannel(pipleChannel);
        Configuration readerPlugin = Configuration.from(f).getConfiguration(Constants.READER_PLUGIN_TAG);
        readerJob.setPluginJobConf(readerPlugin);
        readerJob.init();
    }


    private void writeinit(String writeClassname){
        writerJob = ClassUtil.instantiate(writeClassname, Writer.class, new Object[]{});
        writerJob.setPipleChannel(pipleChannel);
        Configuration writerPlugin = Configuration.from(f).getConfiguration(Constants.WRITE_PLUGIN_TAG);
        writerJob.setPluginJobConf(writerPlugin);
        writerJob.init();
    }

    private void builderReaderJob() {
          Thread thread = new Thread(()->{
              readerJob.readerLine();
         });
        thread.start();

    }

    private void builderWriterJob() {
        Thread thread = new Thread(()->{
        writerJob.writerLine();
        });
        thread.start();
    }

    @Override
    public void stop() {
        readerJob.destroy();
        writerJob.destroy();
        writeflage.set(false);
    }



}
